#include "thread_pool.h"

#include <malloc.h>
#include <mqueue.h>
#include <pthread.h>

#include "log.h"

#define thread_pool_mqueue_name ("/thread_pool_mqueue_name")

typedef struct {
	void *(*function)(void *pv);
	void *argument;
} task;

static void *run(void *pv);

static uint32_t _capacity = 0;
static pthread_t *threads = NULL;
static mqd_t mqd = 0;

int32_t thread_pool_create(uint32_t capacity) {
	struct mq_attr attr = {
		.mq_maxmsg = 10,
		.mq_msgsize = sizeof(task),
	};
	mqd = mq_open(thread_pool_mqueue_name, O_RDWR | O_CREAT, 0644, &attr);
	if (mqd == -1) {
		log_error("创建队列失败\n");
		goto mq_open_fail;
	}
	_capacity = capacity;
	threads = calloc(_capacity, sizeof(pthread_t));
	if (!threads) {
		log_error("申请线程空间失败\n");
		goto threads_calloc_fail;
	}
	uint32_t _i = 0;
	for (uint32_t i = 0; i < _capacity; ++i) {
		if (pthread_create(&threads[i], NULL, run, NULL) != 0) {
			log_error("线程创建失败\n");
			_i = i;
			goto pthread_create_fail;
		}
	}
	return 0;
pthread_create_fail:
	for (uint32_t i = 0; i < _i; ++i) {
		pthread_cancel(threads[i]);
		pthread_join(threads[i], NULL);
	}
	free(threads);
threads_calloc_fail:
	mq_unlink(thread_pool_mqueue_name);
mq_open_fail:
	return -1;
}
int32_t thread_pool_destory(void) {
	for (uint32_t i = 0; i < _capacity; ++i) {
		pthread_cancel(threads[i]);
		pthread_join(threads[i], NULL);
	}
	free(threads);
	mq_unlink(thread_pool_mqueue_name);
	return 0;
}
int32_t thread_pool_run(void *(*function)(void *pv), void *argument) {
	task task = {.function = function, .argument = argument};
	return mq_send(mqd, (char *)&task, sizeof(task), 0);
}

static void *run(void *pv) {
	(void)pv;
	task task = {
		.function = NULL,
		.argument = NULL,
	};
	while (1) {
		if (mq_receive(mqd, (char *)&task, sizeof(task), 0) == -1) {
			log_error("线程接受任务失败\n");
			continue;
		}
		task.function(task.argument);
	}
	return NULL;
}